package com.atguigu.flink.chapter10.collection;

import org.apache.flink.runtime.state.storage.JobManagerCheckpointStorage;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2023/6/29 13:56
 */
public class CollectionDemo1 {
    public static void main(String[] args) {
      
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());
    
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
    
        tEnv.executeSql("create table t1(" +
                            " id string, " +
                            " name string " +
                            ")with(" +
                            "  'connector' = 'kafka', " +
                            "  'topic' = 't1', " +
                            "  'properties.bootstrap.servers' = 'hadoop162:9092', " +
                            "  'properties.group.id' = 'atguigu', " +
                            "  'scan.startup.mode' = 'latest-offset', " +
                            "  'format' = 'csv' " +
                            ")");
    
        tEnv.executeSql("create table t2(" +
                            " id string, " +
                            " nam string " +
                            ")with(" +
                            "  'connector' = 'kafka', " +
                            "  'topic' = 't2', " +
                            "  'properties.bootstrap.servers' = 'hadoop162:9092', " +
                            "  'properties.group.id' = 'atguigu', " +
                            "  'scan.startup.mode' = 'latest-offset', " +
                            "  'format' = 'csv' " +
                            ")");
        
       
//        tEnv.sqlQuery("select * from t1 union  select * from t2").execute().print();
//        tEnv.sqlQuery("select * from t1 Intersect  select * from t2").execute().print();
        tEnv.sqlQuery("select * from t1 except select * from t2").execute().print();
        
    }
}
